package hbase;


import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Hbase 配置，操作
 *
 * @author rikun
 */
public class HbaseOption {

    /**
     * Hbase Connection
     */
    private final Connection con;

    /**
     * Hbase 链接地址配置
     *
     * @param hostAddress String
     */
    public HbaseOption(String hostAddress) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", hostAddress);
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        this.con = ConnectionFactory.createConnection(conf);
        System.out.println("Hbase 已连接");
    }

    /**
     * Hbase 链接地址配置
     *
     * @param hostAddress String
     * @param port String
     */
    public HbaseOption(String hostAddress, String port) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", hostAddress);
        conf.set("hbase.zookeeper.property.clientPort", port);
        this.con = ConnectionFactory.createConnection(conf);
        System.out.println("Hbase 已连接");
    }

    /**
     * 创建表
     *
     * @param name String
     * @throws IOException
     */
    public void hbaseTableCreate(String name) throws IOException {

        String nameSpace = "";
        if (StringUtils.isNotBlank(name)) {
            // 空间名
            String[] temp = name.split(":");
            nameSpace = temp[0];
        }
        else {
            System.out.println("表名不能为空");
        }
        // 创建命名空间
        createNameSpace(nameSpace);
        // 建表
        createTable(name);
    }

    /**
     * 创建命名空间
     *
     * @param nameSpace String
     * @throws IOException
     */
    private void createNameSpace(String nameSpace) throws IOException {
        // nameSpace 不能为空
        if (!StringUtils.isNotEmpty(nameSpace)) {
            nameSpace = "rikun";
        }
        // 获取操作权限
        Admin admin = this.con.getAdmin();
        try{
            admin.getNamespaceDescriptor(nameSpace);
        }catch(NamespaceNotFoundException e){
            //若发生特定的异常，即找不到命名空间，则创建命名空间
            NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
            admin.createNamespace(namespaceDescriptor);
        }
    }

    /**
     * 创建表
     *
     * @param tableName String
     * @throws IOException
     */
    private void createTable(String tableName) throws IOException {
        // 创建表管理类
        Admin admin = this.con.getAdmin();
        // 创建表描述类
        TableName table = TableName.valueOf(tableName);
        if (!admin.tableExists(table)) {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(table);
            hTableDescriptor.addFamily(new HColumnDescriptor("name"));
            hTableDescriptor.addFamily(new HColumnDescriptor("info"));
            hTableDescriptor.addFamily(new HColumnDescriptor("score"));
            // 创建表
            admin.createTable(hTableDescriptor);
            System.out.println("创建表成功！");
        }
        else {
            System.out.println(tableName + " 表已存在");
        }
    }

    /**
     * 插入数据
     *
     * @param tableName String
     * @param data List<Put>
     * @throws IOException
     */
    public void putData(String tableName, List<Put> data) throws IOException {
        // 获取表对象
        Table table = this.con.getTable(TableName.valueOf(tableName));
        // 插入数据
        table.put(data);
        table.close();
    }

    /**
     * 根据 rowKey 删除对应的数据
     *
     * @param tableName String
     * @param rowKey String
     * @throws IOException
     */
    public void deleteDataByRowKey(String tableName, String rowKey) throws IOException {
        // 获取表的对象
        Table table = this.con.getTable(TableName.valueOf(tableName));
        //构建Delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        //删除操作
        table.delete(delete);
        //关闭资源
        table.close();
    }

    /**
     * 根据条件查询
     *
     * @param tableName String
     * @param filters String
     * @return result List<StudentInfo>
     * @throws IOException
     */
    public List<StudentInfo> queryDataByFilter(String tableName, List<Filter> filters) throws IOException {
        // 获取表的对象
        Table table = this.con.getTable(TableName.valueOf(tableName));

        // 查询条件
        Scan scan = new Scan();
        FilterList conditions = new FilterList(filters);
        scan.setFilter(conditions);

        List<StudentInfo> result = new ArrayList<>();
        StudentInfo info = null;

        // 数据拼装
        for (Result r: table.getScanner(scan)) {
            info = new StudentInfo();
            for (Cell cell: r.listCells()) {
                String qualifier = Bytes.toString(cell.getQualifierArray(),
                                                  cell.getQualifierOffset(),
                                                  cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(),
                                              cell.getValueOffset(),
                                              cell.getValueLength());
                if (StringUtils.isEmpty(qualifier)) {
                    info.setName(value);
                }
                else if ("student_id".equals(qualifier)) {
                    info.setStudentId(value);
                }
                else if ("class".equals(qualifier)) {
                    info.setClassId(value);
                }
                else if ("programming".equals(qualifier)) {
                    info.setProgramming(value);
                }
                else if ("understanding".equals(qualifier)) {
                    info.setUnderstanding(value);
                }
            }
            result.add(info);
        }
        return result;
    }





}
